package com.cw.recommend.log

import com.cw.recommend.common.constant._
import org.apache.kafka.streams.processor.{Processor, ProcessorContext, ProcessorSupplier, TopologyBuilder}
import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}

import java.util.Properties

object LogProcess {
  def main(args: Array[String]): Unit = {
    val brokers = KAFKA_BROKERS
    val zookeepers = ZOO_KEEPERS

    // 定义输入和输出的topic
    val from = LOG_TOPIC;
    val to = RECOMMENDER_TOPIC

    // 定义kafka streaming的配置
    val settings = new Properties();
    settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "logFilter");
    settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
    settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, zookeepers);

    val config = new StreamsConfig(settings);

    val builder = new TopologyBuilder();

    builder.addSource("SOURCE", from)
      .addProcessor(
        "PROCESS",
        new ProcessorSupplier[Array[Byte], Array[Byte]] {
          override def get(): Processor[Array[Byte], Array[Byte]] = new Process
        },
        "SOURCE")
      .addSink("SINK", to, "PROCESS");

    val streams = new KafkaStreams(builder, config);
    streams.start();

  }
}


class Process extends Processor[Array[Byte], Array[Byte]] {
  var ctx = Option.empty[ProcessorContext]

  override def init(processorContext: ProcessorContext): Unit = ctx = Some(processorContext)

  override def process(k: Array[Byte], v: Array[Byte]): Unit = {
    val data = new String(v)
    if (data contains "PRODUCT_RATING_PREFIX:") {
      val rating = data.split("PRODUCT_RATING_PREFIX:")(1).trim
      ctx.foreach(_.forward("logProcessor".getBytes(), rating.getBytes))
    }
  }

  override def punctuate(l: Long): Unit = ()

  override def close(): Unit = ()

}